package com.behavioranalysis.flinkprogram.flink.ad;

import com.behavioranalysis.flinkprogram.dao.IAdUserClickCountDAO;
import com.behavioranalysis.flinkprogram.dao.factory.DAOFactory;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

/**
 * <h3>flinkprogram</h3>
 * <p>${description}</p>
 * Created by yang on 20-2-14 下午8:39
 * updated by yang on 20-2-14 下午8:39
 */
public class DailyUserAdClickCountProcess extends
        ProcessWindowFunction<Tuple6<Long, String, String, String, String, String>, Tuple4<Long, Long, String, Long>, String, TimeWindow> {
    private static final long serialVersionUID = 1L;
    private OutputTag<Long> blacklistUserid;

    public DailyUserAdClickCountProcess(OutputTag<Long> blacklistUserid) {
        this.blacklistUserid = blacklistUserid;
    }

    @Override
    public void process(String key, Context context, Iterable<Tuple6<Long, String, String, String, String, String>> elements, Collector<Tuple4<Long, Long, String, Long>> out) throws Exception {
        // 相同的key 的tuple会进到这里面，
        // 那么我们就可以将此窗口中此key的count统计出来存入数据库 <yyyyMMdd_userid_adid, count>
        // 都是一个Key,那么要出如数据库的就是Key,elements的大小
        // 在这里存数据的话，也是每个key需连接一次数据库，
        //System.out.println("此key是：" + key);

        // 2. 这里还需要执行第二个，查询这一天，这一用户，对这一广告的点击量是否已经超过100
        // 这里刚好是按key分组的，我们就是要用key进行查询
        // date, userid, adid
        // count >= 100 将userid返回 拉入黑名单
        // 这个主流我们要用来addSink到数据库
        // 所以将黑名单用户做成侧面输出流，然后在外面进行去重，保存到数据库。
        String[] keySplited = key.split("_");
        // yyyyMMdd -> yyyy-MM-dd
        SimpleDateFormat format =
                new SimpleDateFormat("yyyyMMdd");
        SimpleDateFormat dateFormat =
                new SimpleDateFormat("yyyy-MM-dd");
        String date = dateFormat.format(format.parse(keySplited[0]));
        long userid = Long.valueOf(keySplited[1]);
        long adid = Long.valueOf(keySplited[2]);

        IAdUserClickCountDAO adUserClickCountDAO = DAOFactory.getAdUserClickCountDAO();
        int clickCount = adUserClickCountDAO.findClickCountByMultiKey(
                date, userid, adid);

        // 到这里就是将每一个窗口中，相同的key聚合成了key counts
        // 1. 返回的是<userid, adid, date,counts>，那我们可以直接将此流保存到数据库
        // 需要插入的插入，需要更新的更新
        // TODO
        out.collect(new Tuple4<>(userid, adid, date, Long.valueOf(IteratorUtils.toList(elements.iterator()).size())));

        // 判断，如果点击量大于等于100，就拉入黑名单
        if (clickCount >= 100) {
            // 将此userid加入黑名单侧面输出流
            context.output(blacklistUserid,
                    userid);
        }
    }
}
